package cn.com.cennavi.services;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;

import cn.com.cennavi.bean.MineLabODBean;
import cn.com.cennavi.config.MineLabODConfig;
import cn.com.cennavi.util.MapUtil;

public class DataScreen implements Function<MineLabODBean, MineLabODBean> {

	/** @Fields serialVersionUID: */

	private static final long serialVersionUID = 1L;
	private SimpleDateFormat sf = new SimpleDateFormat("yyyyMMddHHmmss");
	private Map<Integer, Set<Integer>> meshCountyMap = new HashMap<Integer, Set<Integer>>();
	private MineLabODConfig mineLabODConfig;
//	SimpleDateFormat sf = new SimpleDateFormat("yyyyMMddHHmmss");
	public DataScreen(Broadcast<Map<Integer, Set<Integer>>> meshCountyMapBroad, MineLabODConfig mineLabODConfig) {
		this.meshCountyMap = meshCountyMapBroad.value();
		this.mineLabODConfig = mineLabODConfig;
	}

	@Override
	public MineLabODBean call(MineLabODBean md) throws Exception {
		// TODO Auto-generated method stub
		MineLabODBean modn = new MineLabODBean();
		modn.oLatitude = md.getOLatitude();
		modn.dLatitude = md.getDLatitude();
		modn.oLongitude = md.getOLongitude();
		modn.dLongitude = md.getDLongitude();
		modn.oTime = md.getOTime();
		modn.dTime = md.getDTime();
		modn.dataSourse = md.dataSourse;
		boolean oisInArea = false;
		switch (mineLabODConfig.spatialFlag) {
		case 0:
			// 按照区县选取
			oisInArea = oisInCountyArea(modn.dLongitude, modn.dLatitude, mineLabODConfig.countyCode, meshCountyMap);
			break;
		case 1:
			// 按照城市选区域
			oisInArea = oisInCityArea(modn.dLongitude, modn.dLatitude, mineLabODConfig.cityCode, meshCountyMap);
			break;
		case 2:
			// 按照省份选区域
			oisInArea = oisInProvinceArea(modn.dLongitude, modn.dLatitude, mineLabODConfig.province, meshCountyMap);
			break;
		case 3:
			// 按照二次网格筛选区域
			oisInArea = oisInMeshArea(modn.dLongitude, modn.dLatitude, mineLabODConfig.meshCode, meshCountyMap);
			break;
		}
		
		Date s = sf.parse(mineLabODConfig.startDate);
		Date e = sf.parse(mineLabODConfig.endDate);
		boolean disInTime = judgeIsInTime(modn.dTime, s, e);
		if (disInTime && oisInArea) {

			return modn;
		}
		return null;

	}

	private boolean oisInMeshArea(double dLongitude, double dLatitude, int meshCode, Map<Integer, Set<Integer>> meshCountyMap2) {
		// TODO Auto-generated method stub
		int meshId = MapUtil.findMesh(dLongitude, dLatitude);
		if (meshId == meshCode) {
			return true;
		}
		return false;
	}

	private boolean oisInProvinceArea(double dLongitude, double dLatitude, int province, Map<Integer, Set<Integer>> meshCountyMap2) {
		// TODO Auto-generated method stub
		int meshId = MapUtil.findMesh(dLongitude, dLatitude);

		Set<Integer> county = meshCountyMap.get(meshId);
		if (county == null) {
			return false;
		}
		for (int countyId : county) {
			if (countyId / 10000 * 10000 == province) {
				return true;
			}
		}
		return false;
	}

	private boolean oisInCityArea(double dLongitude, double dLatitude, int cityCode, Map<Integer, Set<Integer>> meshCountyMap) {
		// TODO Auto-generated method stub

		int meshId = MapUtil.findMesh(dLongitude, dLatitude);
		System.out.println("meshId:" + meshId);
		Set<Integer> county = meshCountyMap.get(meshId);
		if (county == null) {
			return false;
		}
		for (int countyId : county) {
			int nowCityCode = 0;
			if (cityCode == 110000 || cityCode == 310000 || cityCode == 120000 || cityCode == 500000) {
				nowCityCode = countyId / 10000 * 10000;
			} else {
				nowCityCode = countyId / 100 * 100;
			}
			if (nowCityCode == cityCode) {
				return true;
			}

		}

		return false;
	}

	private boolean oisInCountyArea(double dLongitude, double dLatitude, int countyCode, Map<Integer, Set<Integer>> meshCountyMap) {
		// TODO Auto-generated method stub
		int meshId = MapUtil.findMesh(dLongitude, dLatitude);
		System.out.println(meshId);
		Set<Integer> county = meshCountyMap.get(meshId);
		if (county == null) {
			return false;
		}
		for (int countyId : county) {
			if (countyId == countyCode) {
				return true;
			}
		}

		return false;
	}

	private boolean judgeIsInTime(long dTime, Date s, Date e) throws ParseException {
		// TODO Auto-generated method stub
		Date nowDate = new Date(dTime * 1000);

		if (nowDate.after(s) && nowDate.before(e)) {
			return true;
		}
		return false;
	}

}
